package flink_p2_sql

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table


object tableApi02_tableApi_socketStream {



  case class Person(id:Int, name:String, age:Int)

  /**
   * @param args
   */
  def main(args: Array[String]): Unit = {



    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv)


    val socketStream: DataStream[String] = streamEnv.socketTextStream("127.0.0.1", 8889)

    val dStream: DataStream[Person] = socketStream.map(data => {
      val arr: Array[String] = data.split(" ")
      new Person(arr(0).toInt, arr(1), arr(2).toInt)
    })


    //
//    val table: Table = tableEnv.fromDataStream(dStream)


    import org.apache.flink.table.api.scala._      //需要导入导入隐式转换
    val t_user: Table = tableEnv.fromDataStream(dStream, 'id, 'name, 'age)

    t_user.printSchema()



    tableEnv.execute("test")
  }
}
